﻿using System;
using System.Diagnostics;
using XcoAppSpaces.Core;
using Microsoft.Ccr.Core;
using System.Threading;
using NUnit.Framework;
using XcoAppSpaces.Contracts.Serialization;
using XcoAppSpaces.Serialization.Binary;
using XcoAppSpaces.Serialization.Json;

namespace test.XcoAppSpaces.Core
{
	[TestFixture]
	public class testPerformance
	{
		private const int msgCount = 10000;
		private const int maxTimeSeconds = 15;

		public static void Main()
		{
			try
			{
				new testPerformance().TestPerformanceTcpJson();
			}
			catch (Exception ex)
			{
				Console.WriteLine("ERROR: " + ex);
			}
			Console.ReadLine();
		}


		[Test]
		public void TestPerformanceTcpBinary()
		{
			TestPerformance<XcoBinarySerializer>("tcp");
		}

		[Test]
		public void TestPerformanceWcfBinary()
		{
			TestPerformance<XcoBinarySerializer>("wcf");
		}

		[Test]
		public void TestPerformanceTcpJson()
		{
			TestPerformance<XcoJsonSerializer>("tcp");
		}

		[Test]
		public void TestPerformanceWcfJson()
		{
			TestPerformance<XcoJsonSerializer>("wcf");
		}


		private void TestPerformance<TSerializer>(string commType) where TSerializer : IXcoSerializer, new()
		{
			using (XcoAppSpace space1 = XcoAppSpace.ConfigureWithConfigString(commType + ".port=8100").UsingService<TSerializer>(),
				space2 = XcoAppSpace.ConfigureWithConfigString(commType + ".port=8101").UsingService<TSerializer>())
			{
				ManualResetEvent evt = new ManualResetEvent(false);
				Stopwatch sw = new Stopwatch();
				int count = 0;
				var port = new Port<int>();
				//Arbiter.Activate(space1.DefaultDispatcherQueue, 
				//	Arbiter.Receive(true, port,
				space1.Receive(port,
				                msg =>
				                	{
				                		Interlocked.Increment(ref count);
										if (count % 1000 == 0)
										{
											Console.WriteLine(count + " @ " + sw.Elapsed.TotalSeconds);
											if (count == msgCount)
												evt.Set();
										}
				                	});
				space1.RunWorker(port);

				var remotePort = space2.ConnectWorker<Port<int>>(Environment.MachineName + ":8100");

				var bytesTransferred = 0;
				space2.Info.CommunicationEvents.OnSendingMessage += args =>
					{
						int length = args.Message.Content.Length + args.Message.ContentInfo.Length;
						Interlocked.Add(ref bytesTransferred, length);
					};

				sw.Start();
				//var sw2 = Stopwatch.StartNew();
				for (int i = 0; i < msgCount; i++)
				{
					remotePort.Post(i);
				}
				//sw2.Stop();
				//Console.WriteLine("Sending " + count + " finished @ " + sw2.Elapsed.TotalSeconds);
				Assert.IsTrue(evt.WaitOne(maxTimeSeconds*1000), "Performance test with " + commType + " transport took more than " + maxTimeSeconds + " Seconds - possible performance leak!?");
				Console.WriteLine("Overall bytes transferred: " + bytesTransferred);

				sw.Stop();
			}
		}

		[Test]
		public void TestPerformanceReqResp_TcpBinary()
		{
			TestPerformanceReqResp<XcoBinarySerializer>("tcp");
		}

		[Serializable]
		class Request
		{
			public Port<int> ResponsePort { get; set; }
		}

		private void TestPerformanceReqResp<TSerializer>(string commType) where TSerializer : IXcoSerializer, new()
		{
			using (XcoAppSpace server = XcoAppSpace.ConfigureWithConfigString(commType + ".port=8100").UsingService<TSerializer>(),
				client = XcoAppSpace.ConfigureWithConfigString(commType + ".port=8101").UsingService<TSerializer>())
			{
				ManualResetEvent evt = new ManualResetEvent(false);
				Stopwatch sw = new Stopwatch();
				int count = 0;
				int countResp = 0;
				var port = new Port<Request>();
				server.Receive(port,
								msg =>
								{
									Interlocked.Increment(ref count);
									msg.ResponsePort.Post(count);
									if (count % 1000 == 0)
									{
										Console.WriteLine(count + " requests @ " + sw.Elapsed.TotalSeconds);
										//if (count == msgCount)
										//	evt.Set();
									}
								});
				server.RunWorker(port);

				var responsePort = client.Receive<int>(msg =>
				{
					Interlocked.Increment(ref countResp);
					if (countResp % 1000 == 0)
					{
						Console.WriteLine(countResp + " responses @ " + sw.Elapsed.TotalSeconds);
						if (countResp == msgCount)
							evt.Set();
					}
				});

				var remotePort = client.ConnectWorker<Port<Request>>(Environment.MachineName + ":8100");

				var bytesTransferred = 0;
				client.Info.CommunicationEvents.OnSendingMessage += args =>
				{
					int length = args.Message.Content.Length + args.Message.ContentInfo.Length;
					Interlocked.Add(ref bytesTransferred, length);
				};

				sw.Start();
				//var sw2 = Stopwatch.StartNew();
				for (int i = 0; i < msgCount; i++)
				{
					remotePort.Post(new Request{ResponsePort = responsePort});
				}
				//sw2.Stop();
				//Console.WriteLine("Sending " + count + " finished @ " + sw2.Elapsed.TotalSeconds);
				Assert.IsTrue(evt.WaitOne(maxTimeSeconds * 1000), "Performance test with " + commType + " transport took more than " + maxTimeSeconds + " Seconds - possible performance leak!?");
				Console.WriteLine("Overall bytes transferred: " + bytesTransferred);

				sw.Stop();
			}
		}

		[Test]
		public void TestConnectWorkerPerformance()
		{
			using (XcoAppSpace client = new XcoAppSpace("tcp.port=9001"),
			                   server = new XcoAppSpace("tcp.port=9002"))
			{
				const int num = 20;

				//run workers
				var sw = Stopwatch.StartNew();
				for (int i = 0; i < num; i++)
				{
					var worker = new TestWorkerContract();
					server.RunWorker(worker, "w" + i);
				}
				sw.Stop();
				Console.WriteLine("Time for running workers: " + sw.ElapsedMilliseconds + "ms");

				sw.Reset();
				sw.Start();
				var connected = 0;
				for (int i = 0; i < num; i++)
				{
					int i1 = i;
					new Thread(() =>
						{
							var adr = "localhost:9002/w" + i1;
							//Console.WriteLine("start " + adr);
							client.ConnectWorker<TestWorkerContract>(adr, "w" + i1);
							Interlocked.Increment(ref connected);
							//Console.WriteLine("finished " + adr);

						}).Start(); //start in own thread so that is doesn't block the threadpool and prevents the server from accepting a new connection...
					//Console.WriteLine(DateTime.Now.Millisecond);
				}
				Thread.Sleep(1000);
				//var result = SpinWait.SpinUntil(() => connected == num, 1000);
				sw.Stop();
				Console.WriteLine("Time for connecting workers: " + sw.ElapsedMilliseconds + "ms");
				Assert.IsTrue(connected == num);
			}
		}

		[Serializable]
		public class TestMsg1
		{
			public int Prop1 { get; set; }
			public string Prop2 { get; set; }
			public string ResponsePort { get; set; }

		}

		[Serializable]
		public class TestMsg2
		{
			public int Prop1 { get; set; }
			public string Prop2 { get; set; }
			public string ResponsePort { get; set; }
		}
		
		public class TestWorkerContract : PortSet<TestMsg1, TestMsg2, Subscribe<TestMsg1>, Unsubscribe<TestMsg1>>
		{
		}
	}
}
